package com.example.emqxclient.client;

import com.example.emqxclient.properties.MqttProperties;
import com.example.emqxclient.properties.QosEnum;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.stereotype.Component;

/**
 * File: EmqClient.java
 * Author: liufudong@szyh-smart.com
 * Date: 2024/8/29
 * Description: EmqClient class.
 */
@Component
@Slf4j
public class EmqClient {
    private IMqttClient mqttClient;
    @Resource
    private MqttProperties mqttProperties;
    @Resource
    private MessageCallback messageCallback;

    @PostConstruct
    public void init() {
        try {
            mqttClient = new MqttClient(mqttProperties.getBrokerUrl(),
                    mqttProperties.getClientId(), new MemoryPersistence());
        } catch (MqttException e) {
            e.printStackTrace();
            log.error("MqttClient初始化失败,brokerurl={},clientId={}",
                    mqttProperties.getBrokerUrl(), mqttProperties.getClientId());
        }
    }

    /**
     * 连接broker
     *
     * @param username
     * @param password
     */
    public void connect(String username, String password) {
        MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
        //创建MQTT连接选项对象--可配置mqtt连接相关选项
        //自动重连
        mqttConnectOptions.setAutomaticReconnect(true);
/**
 * 设置为true后意味着：客户端断开连接后emq不保留会话保留会话，否则会产生订阅共享队列的存活
 客户端收不到消息的情况
 * 因为断开的连接还被保留的话，emq会将队列中的消息负载到断开但还保留的客户端，导致存活的客户
 端收不到消息
 * 解决该问题有两种方案:1.连接断开后不要保持；2.保证每个客户端有固定的clientId
 */
        mqttConnectOptions.setCleanSession(true);
        mqttConnectOptions.setUserName(username);
        mqttConnectOptions.setPassword(password.toCharArray());
//设置mqtt消息回调
        mqttClient.setCallback(messageCallback);
        try {
            mqttClient.connect(mqttConnectOptions);
            // 订阅客户端连接状态

            // Keep the application running to receive messages
//            Thread.sleep(Long.MAX_VALUE);
        } catch (MqttException e) {
            e.printStackTrace();
            log.error("连接mqtt broker失败,失败原因:{}", e.getMessage());
        }
    }
    /**
     * 发布
     * @param topic
     * @param msg
     */
    public void publish(String topic, String msg, QosEnum qos, boolean retain){
        MqttMessage mqttMessage=new MqttMessage();
        mqttMessage.setPayload(msg.getBytes());
        mqttMessage.setQos(qos.value());
        mqttMessage.setRetained(retain);
        try {
            mqttClient.publish(topic,mqttMessage);
        } catch (MqttException e) {
            e.printStackTrace();
            log.error("mqtt publish失败,失败原因:{}", e.getMessage());
        }
    }
    /**
     * 订阅
     *
     * @param topicFilter
     * @return
     */
    public void subscribe(String topicFilter, QosEnum qos) {
        try {
            mqttClient.subscribe(topicFilter, qos.value());
        } catch (MqttException e) {
            e.printStackTrace();
            log.error("mqtt subscribe失败,失败原因:{}", e.getMessage());
        }
    }

    /**
     * 断开连接
     */
    @PreDestroy
    public void onDestroy() {
        try {
            mqttClient.disconnect();
        } catch (MqttException e) {
            e.printStackTrace();
            log.error("mqtt disconnect失败,失败原因:{}", e.getMessage());
        }
    }
}
